package com.hudson.store.index;

import com.hudson.core.base.BaseComponent;
import com.hudson.store.util.MappedFile;

import java.nio.MappedByteBuffer;
import java.util.HashMap;
import java.util.Map;

public class IndexFile extends BaseComponent {

    private int hashSlotNum = 500_000;

    private int headerSize = 40;

    private static int invalidIndex = 0;

    private MappedFile mappedFile;
    //索引文件头
    private IndexHeader indexHeader;

    public IndexFile(String fileName) {
        this.mappedFile = new MappedFile(fileName, 40 + hashSlotNum * 4L + 2000_0000 * 24);
        this.indexHeader = new IndexHeader(mappedFile.getMappedByteBuffer());
    }

    //4 -> keyHash  8 -> offset  4 -> 时间差值  4 -> 上一条记录的索引
    private final int indexSize = 24;

    public void putKey(String key, long offset, int messageSize, long storeTimeStamp) {
        int hashcode = getHashCode(key);
        int hashSlot = hashcode % hashSlotNum;
        int slotOffset = headerSize + hashSlot * 4;
        MappedByteBuffer byteBuffer = mappedFile.getMappedByteBuffer();

        try {
            int slotValue = byteBuffer.getInt(slotOffset);
            if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCountAtomic()) {
                slotValue = invalidIndex;
            }

            int indexOffset = headerSize + 4 * hashSlotNum + indexSize * this.indexHeader.getIndexCountAtomic();

            long timeDiff = storeTimeStamp - this.indexHeader.getStartTimeStampAtomic();

            timeDiff = timeDiff / 1000;

            if (this.indexHeader.getStartTimeStampAtomic() <= 0) {
                timeDiff = 0;
            } else if (timeDiff > Integer.MAX_VALUE) {
                timeDiff = Integer.MAX_VALUE;
            } else if (timeDiff < 0) {
                timeDiff = 0;
            }

            //hash值
            byteBuffer.putInt(indexOffset, hashcode);
            //commitLog中的索引值
            byteBuffer.putLong(indexOffset + 4, offset);
            //message的大小
            byteBuffer.putInt(indexOffset + 4 + 8, messageSize);
            //与文件创建时间差值
            byteBuffer.putInt(indexOffset + 4 + 8 + 4, (int) timeDiff);
            //上一条数据的index位置
            byteBuffer.putInt(indexOffset + 4 + 8 + 4 + 4, slotValue);

            byteBuffer.putInt(slotOffset, this.indexHeader.getIndexCountAtomic());

            if (this.indexHeader.getIndexCountAtomic() <= 1) {
                this.indexHeader.setStartOffsetLimitAtomic(offset);
                this.indexHeader.setStartTimeStampAtomic(storeTimeStamp);
            }

            if (invalidIndex == slotValue) {
                this.indexHeader.incHashSlotCountAtomic();
            }
            this.indexHeader.incIndexCountAtomic();
            this.indexHeader.updateEndOffsetLimitAtomic(offset);
            this.indexHeader.updateEndTimeStampAtomic(storeTimeStamp);

            mappedFile.flush();
            indexHeader.flush();
        } catch (Exception e) {
            log.error("putKey exception, Key: {} KeyHashCode: {}", key, key.hashCode(), e);
        }
    }

    public Map<Long, Integer> getIndex(String key, int size) {
        Map<Long, Integer> result = new HashMap<>();

        int hashcode = getHashCode(key);
        int hashSlot = hashcode % hashSlotNum;
        int slotOffset = headerSize + hashSlot * 4;
        MappedByteBuffer byteBuffer = mappedFile.getMappedByteBuffer();

        try {
            int slotValue = byteBuffer.getInt(slotOffset);
            if (slotValue <= invalidIndex || slotValue > this.indexHeader.getIndexCountAtomic()) {
                return result;
            }

            int indexOffset = headerSize + 4 * hashSlotNum + indexSize * slotValue;
            boolean flag = true;

            while (flag) {
                int code = byteBuffer.getInt(indexOffset);
                if (hashcode == code) {
                    result.put(byteBuffer.getLong(indexOffset + 4), byteBuffer.getInt(indexOffset + 4 + 8));
                }
                int nextSlotOffset = byteBuffer.getInt(indexOffset + 4 + 8 + 4 + 4);
                if (nextSlotOffset <= invalidIndex || nextSlotOffset > this.indexHeader.getIndexCountAtomic()) {
                    flag = false;
                } else {
                    indexOffset = headerSize + 4 * hashSlotNum + indexSize * nextSlotOffset;
                }

                if (result.size() == size) {
                    flag = false;
                }
            }
        } catch (Exception e) {
            log.error("putKey exception, Key: {} KeyHashCode: {}", key, key.hashCode(), e);
        }

        return result;
    }

    private int getHashCode(String key) {
        int hashCode = key.hashCode();
        return Math.abs(hashCode);
    }
}
